package com.ptkc.core.canal;

import com.alibaba.otter.canal.client.CanalConnector;
import com.alibaba.otter.canal.client.CanalConnectors;
import com.alibaba.otter.canal.common.utils.AddressUtils;
import com.alibaba.otter.canal.protocol.CanalEntry;
import com.alibaba.otter.canal.protocol.Message;
import com.alibaba.otter.canal.protocol.exception.CanalClientException;
import com.google.protobuf.InvalidProtocolBufferException;
import com.ptkc.core.canal.config.CanalProperties;
import com.ptkc.core.canal.handler.EventHandler;
import com.ptkc.core.canal.handler.EventHandlerSuppor;
import com.ptkc.utils.StringUtil;
import lombok.extern.slf4j.Slf4j;
import org.elasticsearch.ElasticsearchStatusException;
import org.elasticsearch.client.ResponseException;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.CommandLineRunner;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.stereotype.Component;

import java.net.InetSocketAddress;
import java.util.*;

/**
 * @author: Mr.Jiao
 * @date: 2018年09月29日
 */
@Component
@EnableConfigurationProperties(CanalProperties.class)
@Slf4j
public class CanalInitRunner implements CommandLineRunner {

    @Autowired
    private CanalProperties canalProperties;

    @Autowired
    private EventHandlerSuppor handlerSuppor;


    public CanalConnector getCanalConn(){
        CanalConnector connector = null;
        if(StringUtil.isNotEmpty(canalProperties.getServer())){
            connector = CanalConnectors.newClusterConnector(canalProperties.getServer(), canalProperties.getDestination(), canalProperties.getUsername(), canalProperties.getPassword());
        }else if(StringUtil.isNotEmpty(canalProperties.getHost()) && canalProperties.getPort() != null){
            connector = CanalConnectors.newSingleConnector(new InetSocketAddress(canalProperties.getHost(), canalProperties.getPort()), canalProperties.getDestination(), canalProperties.getUsername(), canalProperties.getPassword());
        }else{
            throw new NullPointerException("can't read com.ptkc.util.canal host:port or server");
        }
        Collection<EventHandler> handlers = handlerSuppor.getHandlers();
        StringBuilder sb = new StringBuilder();
        handlers.forEach(handler -> {
            sb.append(handler.getDatabase()).append(".").append(handler.getTable()).append(",");
        });
        connector.connect();
        //订阅 监控的 数据库.表  指定filter，格式 {database}.{table}，这里不做过滤，过滤操作留给用户
        connector.subscribe(sb.toString());
        //一次取3条
        // 回滚寻找上次中断的位置
        connector.rollback();
        return connector;
    }

    @Override
    public void run(String... args) throws Exception {
        CanalConnector connector = this.getCanalConn();
        while(true) {
            Message message = connector.getWithoutAck(canalProperties.getBatchSize());
            long batchId = message.getId();
            try {
                List<CanalEntry.Entry> entries = message.getEntries();
                if (batchId != -1 && entries.size() > 0) {
                    this.handleCanalEvent(entries);
                    connector.ack(batchId);
                } else {
                    connector.ack(batchId);
                    Thread.sleep(1000);
                }
            } catch (Exception e) {
                log.error("Failed to send monitoring handler! batchId is rollback ,batchId = " + batchId, e);
//                connector.rollback(batchId);
            }
        }
    }

    private void handleCanalEvent(List<CanalEntry.Entry> entries) {
        entries.forEach(entry -> {
            try {
                CanalEntry.RowChange change = CanalEntry.RowChange.parseFrom(entry.getStoreValue());
                handlerSuppor.handle(entry, change);
            } catch (InvalidProtocolBufferException e) {
                log.error("canalEntry_parser_error,read RowChange By Entry is fail!", e);
            } catch (ElasticsearchStatusException e){
                log.error("canalEntry_parser_error,read RowChange By Entry is fail!", e);
            } catch (CanalClientException e){
                log.error("canalEntry_parser_error,read RowChange By Entry is fail!", e);
            } catch (Throwable e){
                log.error("canalEntry_parser_error,read RowChange By Entry is fail!", e);
            }
        });
    }

}
